package com.bw.day0627;



import com.bw.utils.MyKafkaUtil;
import com.bw.utils.MysqlUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;


public class DwdTraderOrderRefund {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv= StreamTableEnvironment.create(env);

        tableEnv.getConfig().getConfiguration();
        tableEnv.executeSql("create table topic_db(\n" +
                " `database` string,\n" +
                " `table` string,\n" +
                " `type` string,\n" +
                " `data` MAP<STRING,STRING>,\n" +
                " `old`  MAP<STRING,STRING>,\n" +
                " `proc_time` as PROCTIME(),\n" +
                " `ts`  string\n" +
                ")"+ MyKafkaUtil.getKafkaDDL("topic_db","order_refund"));
//        tableEnv.executeSql(MysqlUtil.getBaseDicLookUpDDL());
        Table table = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['user_id'] user_id,\n" +
                "data['order_id'] order_id,\n" +
                "data['sku_id'] sku_id,\n" +
                "data['refund_type'] refund_type,\n" +
                "data['refund_num'] refund_num,\n" +
                "data['refund_amount'] refund_amount,\n" +
                "data['refund_reason_type'] refund_reason_type,\n" +
                "data['refund_reason_txt'] refund_reason_txt,\n" +
                "data['refund_status'] refund_status,\n" +
                "data['create_time'] create_time,\n" +
                "proc_time,\n" +
                "ts\n" +
                " from topic_db \n" +
                "where `table`='order_refund_info' \n" +
                "and  `type`='insert'");
        tableEnv.createTemporaryView("order_refund_info",table);
//        tableEnv.toAppendStream(table, Row.class).print(">>>>>");

//        tableEnv.executeSql(MysqlUtil.getBaseDicLookUpDDL());
        Table table1 = tableEnv.sqlQuery("select \n" +
                "data['id'] id ,\n" +
                "data['province_id'] province_id ,\n" +
                "`old`\n" +
                " from  topic_db where `table` ='order_info'\n" +
                "and `type` ='update' " +
                "and data['order_status']='1005'\n" +
                "and `old`['order_status'] is not null");
        tableEnv.createTemporaryView("order_info_refund", table1);
//        tableEnv.toAppendStream(table1, Row.class).print();

        tableEnv.executeSql(MysqlUtil.getBaseDicLookUpDDL());
        Table table2 = tableEnv.sqlQuery("select\n" +
                "ri.id                 ," +
                "ri.user_id            ," +
                "ri.order_id           ," +
                "ri.sku_id             ," +
                "ri.refund_type        ," +
                "ri.refund_num         ," +
                "ri.refund_amount      ," +
                "ri.refund_reason_type ," +
                "ri.refund_reason_txt  ," +
                "ri.refund_status ,\n" +
                "date_format(ri.create_time,'yyyy-MM-dd') date_id,\n" +
                "oi.province_id,\n" +
                "type_dic.dic_name,\n" +
                "reason_dic.dic_name,\n" +
                "ri.ts,\n" +
                "current_row_timestamp() row_op_ts\n" +
                "from order_refund_info ri join order_info_refund oi \n" +
                "on ri.order_id =oi.id join base_dic for system_time as of ri.proc_time as type_dic\n" +
                "on ri.refund_type = type_dic.dic_code join base_dic for system_time as of   ri.proc_time as reason_dic\n" +
                "on ri.refund_reason_type =reason_dic.dic_code");
        tableEnv.createTemporaryView("result_table", table2);
//        tableEnv.toAppendStream(table2, Row.class).print();

        tableEnv.executeSql("create table dwd_trade_order_refund( " +
                "id string, " +
                "user_id string, " +
                "order_id string, " +
                "sku_id string, " +
                "province_id string, " +
                "date_id string, " +
                "create_time string, " +
                "refund_type_code string, " +
                "refund_type_name string, " +
                "refund_reason_type_code string, " +
                "refund_reason_type_name string, " +
                "refund_reason_txt string, " +
                "refund_num string, " +
                "refund_amount string, " +
                "ts string, " +
                "row_op_ts timestamp_ltz(3) " +
                ")" + MyKafkaUtil.getKafkaSinkDDL("dwd_trade_order_refund"));

        tableEnv.executeSql("insert into dwd_trade_order_refund select * from result_table");

//        env.execute();
    }
}
